package com.guass.nettyprotostuff.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;


import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * @author guass
 * @date 2021年12月22日 17:37
 */
public class RingBufferWorkerPoolFactory {

    private static class SingletonHolder {
        static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
    }

    private RingBufferWorkerPoolFactory(){

    }

    public static RingBufferWorkerPoolFactory getInstance() {
        return SingletonHolder.instance;
    }

    private  Map<String, CommandProducer> producers = new ConcurrentHashMap<>();

    private  Map<String, CommandConsumer> consumers = new ConcurrentHashMap<>();

    private RingBuffer<TranslatorDataWapper> ringBuffer;

    private SequenceBarrier sequenceBarrier;

    private WorkerPool<TranslatorDataWapper> workerPool;

    public void initAndStart(ProducerType type,
                             int bufferSize, WaitStrategy waitStrategy, CommandConsumer[] commandConsumers, Executor executor) {
        //1. 构建ringBuffer对象
        this.ringBuffer = RingBuffer.create(type,
                () -> new TranslatorDataWapper(),
                bufferSize,
                waitStrategy);
        //2.设置序号栅栏
        this.sequenceBarrier = this.ringBuffer.newBarrier();
        //3.设置工作池
        this.workerPool = new WorkerPool<>(this.ringBuffer,
                this.sequenceBarrier,
                new EventExceptionHandler(), commandConsumers);
        //4 把所构建的消费者置入池中
        if (commandConsumers != null && commandConsumers.length > 0) {
            for (CommandConsumer commandConsumer : commandConsumers) {
                this.consumers.put(commandConsumer.getConsumerId(),commandConsumer);
            }
        }
        //5 添加我们的sequences
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());
        //6 启动我们的工作池
//        this.workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()/2));
        this.workerPool.start(executor);
    }

    public CommandProducer getMessageProducer(String producerId){
        CommandProducer messageProducer = this.producers.get(producerId);
        if(null == messageProducer) {
            messageProducer = new CommandProducer(producerId, this.ringBuffer);
            this.producers.put(producerId, messageProducer);
        }
        return messageProducer;
    }


    /**
     * 异常静态类
     * @author Alienware
     *
     */
    static class EventExceptionHandler implements ExceptionHandler<TranslatorDataWapper> {
        @Override
        public void handleEventException(Throwable ex, long sequence, TranslatorDataWapper event) {
        }

        @Override
        public void handleOnStartException(Throwable ex) {
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
        }
    }

}
